package com.yang965.producer.admin;

import org.apache.kafka.clients.admin.*;
import org.apache.kafka.common.config.ConfigResource;

import java.util.Collections;
import java.util.Properties;
import java.util.concurrent.ExecutionException;

public class Admin {
    public static final String bootstrapServers = "192.168.137.33:9092";


    public static void main(String[] args) {
        Properties properties = new Properties();
        properties.put(AdminClientConfig.BOOTSTRAP_SERVERS_CONFIG, bootstrapServers);
        properties.put(AdminClientConfig.REQUEST_TIMEOUT_MS_CONFIG, 300000);
        AdminClient client = AdminClient.create(properties);

        String topic = "topic-demo-1";
        // 创建主题
        createTopic(client, topic);
        // 查看主题
        describe(client, topic);

        client.close();
    }

    private static void describe(AdminClient client, String topic) {
        ConfigResource configResource = new ConfigResource(ConfigResource.Type.TOPIC, topic);
        DescribeConfigsResult describeConfigsResult = client.describeConfigs(Collections.singleton(configResource));
        Config config = null;
        try {
            config = describeConfigsResult.all().get().get(configResource);
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
        System.out.println("主题信息：" + config.toString());
    }

    private static void createTopic(AdminClient client, String topic) {
        // 分区数3 副本因子3
        NewTopic newTopic = new NewTopic(topic, 3, (short) 3);
        CreateTopicsResult topics = client.createTopics(Collections.singleton(newTopic));
        try {
            topics.all().get();
        } catch (InterruptedException e) {
            e.printStackTrace();
        } catch (ExecutionException e) {
            e.printStackTrace();
        }
    }

}
